#include "conn_node_client.h"
#include "time_helper.h"
#include "game_event.h"
#include "websocket.h"
#include "tea.h"
#include "flow_record.h"
#include "kafka_thread.h"

#include <errno.h>
#include <string.h>
#include <assert.h>
#include <stdlib.h>
#include <stdint.h>
#include <unistd.h>

conn_node_client::conn_node_client()
{
}

conn_node_client::~conn_node_client()
{
	send_player_exit();	
}

int conn_node_client::decode_and_check_crc(PROTO_HEAD *head)
{
/*
	int len = ENDION_FUNC_4(head->len) - sizeof(uint16_t);
	char *p = (char *)&head->msg_id;
	int pos = 0;
	while (pos + 8 <= len)
	{
		sg_decrypt((uint32_t *)&p[pos]);
		pos += 8;
	}
		//check seq

		//心跳包不检测
	if (ENDION_FUNC_2(head->msg_id) == 0)
	{
		++seq;
		return (0);
	}

	if (ENDION_FUNC_2(head->msg_id) == LOGIN_REQUEST)
		seq = ENDION_FUNC_2(head->seq);

	if (seq != ENDION_FUNC_2(head->seq))
	{
		LOG_ERR("%s %d: check seq fail, msg id = %d seq[%d] head->seq[%d]", __PRETTY_FUNCTION__, __LINE__, ENDION_FUNC_2(head->msg_id), seq, ENDION_FUNC_2(head->seq));
		return (-1);
	}

	uint32_t crc32 = crc32_long((u_char*)head->data, ENDION_FUNC_4(head->len) - sizeof(PROTO_HEAD));
	if (head->crc != ENDION_FUNC_4(crc32))
	{
		LOG_ERR("%s %d: check crc fail, msg id = %d seq[%d] head->seq[%d]", __PRETTY_FUNCTION__, __LINE__, ENDION_FUNC_2(head->msg_id), seq, ENDION_FUNC_2(head->seq));
		return (-10);
	}
	++seq;
*/
	return (0);
}

int conn_node_client::send_hello_resp()
{
	struct timeval tv;
	time_helper::get_current_time(&tv);
//	tv.tv_sec;
	return (0);
}

void conn_node_client::memmove_data()
{
	int len = buf_size();
	if (len == 0)
		return;
	memmove(&buf[0], buf_head(), len);
	pos_begin = 0;
	pos_end = len;

	LOG_DEBUG("%s %d: memmove happened, len = %d", __PRETTY_FUNCTION__, fd, len);
}
void conn_node_client::remove_buflen(int len)
{
	assert(len <= buf_size());
	if (len == buf_size())
	{
		pos_begin = pos_end = 0;
		return;
	}
	pos_begin += len;
	return;
}

int conn_node_client::respond_websocket_request()
{
//	ws_req_t ws_req;
//	parse_websocket_request((char *)buf_head(), &ws_req); //parse request
//	print_websocket_request(&ws_req);
	// 	//TODO
	// 	//check if it is a websocket request
	// 	//if (!valid) {
	// 	//	ws_serve_exit(conn);
	// 	//	return;
	// 	//}

	char *key = get_websocket_request_key((char *)buf_head());
	if (!key)
	{
		return -1;
	}

//	LOG_DEBUG("%s: read key %s", __FUNCTION__, key);
	
	int len;
	char *resp = generate_websocket_response(key, &len); //generate response
	if (resp)
	{
//		LOG_DEBUG("%s: gen resp %s, len = %d", __FUNCTION__, resp, len);
		send_one_buffer(this, resp, len);
	}
	return (0);
}

extern int send_one_buffer(conn_node_base *node, uint8_t fin, uint8_t opcode, char *buffer, uint32_t len);
void conn_node_client::on_recv_frame()
{
	if ((uint64_t)buf_size() < frame.payload_len)
		return;
	if (flag & NODE_DISCONNECTING)
		return;

	// char t[10] = "a,b,c,d\n";
	// send_one_buffer(this, 1, 2, (char *)buf_head(), frame.payload_len);
	// send_one_buffer(this, 1, 2, t, strlen(t) + 1);
	// return;
	
	uint16_t t_cmd = *((uint16_t *)(buf_head()));
	uint16_t t_seq = *((uint16_t *)(buf_head() + 2));
	LOG_DEBUG("%s: cmd = %u, seq = %u, player_id = %lu", __FUNCTION__, t_cmd, t_seq, player_id);
	UNUSED(t_seq);
	if (player_id)
	{
		switch (t_cmd)
		{
			// case MSGID_HEART:
			// 	send_one_buffer(this, 1, 2, (char *)&t_cmd, sizeof(t_cmd));
			// 	break;
			// case MSGID_FBPAY_REQ:
			// 	transfer_to_fbpayserver();
			// 	break;
			// default:
			// 	transfer_to_gameserver();
			// 	break;
		}
	}
	else
	{
		// transfer_to_loginserver();
	}
		
	// if (write_to_shm_ipc(ipc_game_wr, buf_head()) != 0)
	// {
	// 	LOG_ERR("%s: shm ipc no mem", __FUNCTION__);
	// 	exit(0);
	// }
			
	
//	Grace__Proto__Msg__Player *player = grace__proto__msg__player__unpack(NULL, buf_size(), buf_head());
//	UNUSED(player);
//	char *data = (char *)buf_head();
//					LOG_DEBUG("%s: get frame step4 fin %d opcode %d, len = %d, data = %s",
//						__FUNCTION__, frame.fin, frame.opcode, frame.payload_len, data);
//					int len = set_frame_head(1, 1, frame.payload_len, debug_send_buf);
//					memcpy(&debug_send_buf[len], data, frame.payload_len);
//					send_one_buffer(1, 1, data, frame.payload_len);
	
}

int conn_node_client::frame_read_cb(int fd)
{
	int ret = recv_from_fd();
	if (ret != 0)
		return ret;

	for (;;)
	{
		int len = buf_size();
		switch (conn_step)
		{
			case 1:
			{
				if (len < 2)
					return (0);
//				LOG_DEBUG("---- STEP 1 ---- len %d", len);
//			char tmp[conn->ntoread];
//			bufferevent_read(bev, tmp, conn->ntoread);
					//parse header
				if (parse_frame_header((const char *)buf_head(), &frame) == 0) {
					// LOG_DEBUG("FIN         = %u", frame.fin);
					// LOG_DEBUG("OPCODE      = %u", frame.opcode);
					// LOG_DEBUG("MASK        = %u", frame.mask);
					// LOG_DEBUG("PAYLOAD_LEN = %lu", frame.payload_len);
						//payload_len is [0, 127]
					if (frame.payload_len <= 125) {
						conn_step = 3;
//					conn->ntoread = 4;
//					bufferevent_setwatermark(bev, EV_READ, conn->ntoread, conn->ntoread);
					} else if (frame.payload_len == 126) {
						conn_step = 2;
//					conn->ntoread = 2;
//					bufferevent_setwatermark(bev, EV_READ, conn->ntoread, conn->ntoread);
					} else if (frame.payload_len == 127) {
						conn_step = 2;
//					conn->ntoread = 8;
//					bufferevent_setwatermark(bev, EV_READ, conn->ntoread, conn->ntoread);
					}
				}
				remove_buflen(2);
					//TODO
					//validate frame header
				if (!is_frame_valid(&frame)) {
					LOG_ERR("%s: not a valid frame", __FUNCTION__);
					return (-1);
				}
				break;
			}

			case 2:
			{
//				LOG_DEBUG("---- STEP 2 ---- len %d", len);
//			char tmp[conn->ntoread];
//			bufferevent_read(bev, tmp, conn->ntoread);
				if (frame.payload_len == 126) {
					if (len < 2)
						return (0);
					frame.payload_len = ntohs(*(uint16_t*)buf_head());
					remove_buflen(2);				
					LOG_DEBUG("PAYLOAD_LEN = %lu", frame.payload_len);
				} else if (frame.payload_len == 127) {
					if (len < 8)
						return (0);
					frame.payload_len = ntohl(*(uint64_t*)buf_head());
					remove_buflen(8);
					LOG_DEBUG("PAYLOAD_LEN = %lu", frame.payload_len);
				}
				conn_step = 3;
//			conn->ntoread = 4;
//			bufferevent_setwatermark(bev, EV_READ, conn->ntoread, conn->ntoread);
				break;
			}

			case 3:
			{
//			char tmp[conn->ntoread];
//			bufferevent_read(bev, tmp, conn->ntoread);
//				LOG_DEBUG("---- STEP 3 ---- len %d", len);

				if (frame.mask)
				{
					if (len < 4)
						return (0);
					memcpy(frame.masking_key, buf_head(), 4);
					remove_buflen(4);
				}
				if (frame.payload_len > 0) {
					conn_step = 4;
//				conn->ntoread = conn->frame->payload_len;
//				bufferevent_setwatermark(bev, EV_READ, conn->ntoread, conn->ntoread);
				} else if (frame.payload_len == 0) {
						/*recv a whole frame*/
//					if (frame.mask == 0) {
							//recv an unmasked frame
//					}
					if (frame.fin == 1 && frame.opcode == 0x8) {
							//0x8 denotes a connection close
//						char resp[32];
//						int len = set_frame_head(1, 8, 0, resp);
						send_one_buffer(this, 1, 8, NULL, 0);
					
							// frame_buffer_t *fb = frame_buffer_new(1, 8, 0, NULL);
							// send_a_frame(conn, fb);
						LOG_DEBUG("send a close frame");
							// frame_buffer_free(fb);
						return (-1);
					} else if (frame.fin == 1 && frame.opcode == 0x9) {
						LOG_DEBUG("get a ping frame");									
							//0x9 denotes a ping
							//TODO
							//make a pong
//						char resp[32];
//						int len = set_frame_head(1, 10, 0, resp);
						send_one_buffer(this, 1, 10, NULL, 0);
					} else {
						LOG_DEBUG("%s: get frame step3 fin %d opcode %d", __FUNCTION__, frame.fin, frame.opcode);
							//todo 添加处理
							//execute custom operation
							// if (conn->frame_recv_cb_unit.cb) {
							// 	websocket_cb cb = conn->frame_recv_cb_unit.cb;
							// 	void *cbarg = conn->frame_recv_cb_unit.cbarg;
							// 	cb(cbarg);
							// }
					}

					conn_step = 1;
					memmove_data();
//				conn->ntoread = 2;
//				bufferevent_setwatermark(bev, EV_READ, conn->ntoread, conn->ntoread);
				}
				break;
			}

			case 4:
			{
//				LOG_DEBUG("---- STEP 4 ---- len %d", len);
				if (frame.payload_len > 0) {
					if (len < (int)frame.payload_len)
						return 0;
						// if (conn->frame->payload_data) {
						// 	delete[] conn->frame->payload_data;
						// 	conn->frame->payload_data = NULL;
						// }
						// conn->frame->payload_data = new char[conn->frame->payload_len];
						// bufferevent_read(bev, conn->frame->payload_data, conn->frame->payload_len);
					if (frame.mask)
						unmask_payload_data((char *)buf_head(), frame.payload_len, frame.masking_key);
				}


					/*recv a whole frame*/
				if (frame.fin == 1 && frame.opcode == 0x8) {
						//0x8 denotes a connection close
						// frame_buffer_t *fb = frame_buffer_new(1, 8, 0, NULL);
						// send_a_frame(conn, fb);
					LOG_DEBUG("send a close frame");
						// frame_buffer_free(fb);
//					char resp[32];
//					int len = set_frame_head(1, 8, 0, resp);
					send_one_buffer(this, 1, 8, NULL, 0);
					return (-1);
//					break;
				} else {
					if (frame.fin == 1 && frame.opcode == 0x9) {
						LOG_DEBUG("get a ping frame");				
							//0x9 denotes a ping
							//TODO
							//make a pong
//						char resp[32];
//						int len = set_frame_head(1, 10, 0, resp);
						send_one_buffer(this, 1, 10, NULL, 0);
					}

					on_recv_frame();
//					char *data = (char *)buf_head();
//					data[frame.payload_len] = '\0';
//					LOG_DEBUG("%s: get frame step4 fin %d opcode %d, len = %d, data = %s",
//						__FUNCTION__, frame.fin, frame.opcode, frame.payload_len, data);
//					int len = set_frame_head(1, 1, frame.payload_len, debug_send_buf);
//					memcpy(&debug_send_buf[len], data, frame.payload_len);
//					send_one_buffer(1, 1, data, frame.payload_len);
						//todo 添加处理
						//execute custom operation
						// if (conn->frame_recv_cb_unit.cb) {
						// 	websocket_cb cb = conn->frame_recv_cb_unit.cb;
						// 	void *cbarg = conn->frame_recv_cb_unit.cbarg;
						// 	cb(cbarg);
						// }
					remove_buflen(frame.payload_len);			
				}


				if (frame.opcode == 0x1) { //0x1 denotes a text frame
				}
				if (frame.opcode == 0x2) { //0x1 denotes a binary frame
				}


				conn_step = 1;
				memmove_data();
//			conn->ntoread = 2;
//			bufferevent_setwatermark(bev, EV_READ, conn->ntoread, conn->ntoread);
				break;
			}

			default:
//				LOG_DEBUG("---- STEP UNKNOWN ----");
				LOG_DEBUG("exit");
				exit(-1);
				break;
		}
	}
	return (0);
}

int conn_node_client::recv_from_fd()
{
	int ret = -1;

	if (ssl)
	{
		ret = SSL_read(ssl, buf_tail(), buf_leave());		
	}
	else
	{
		ret = recv(fd, buf_tail(), buf_leave(), 0);
	}

//	LOG_DEBUG("%s: ret = %d", __FUNCTION__, ret);
	
	if (ret == 0) {
		LOG_INFO("%s %d %d: recv ret [%d] err [%d] buf[%p] pos_begin[%d] pos_end[%d]", __PRETTY_FUNCTION__, __LINE__, fd, ret, errno, buf, pos_begin, pos_end);
		return (-1);
	}
	else if (ret < 0) {
		if (errno != EAGAIN && errno != EINTR) {
			LOG_ERR("%s %d %d: recv ret [%d] err [%d] buf[%p] pos_begin[%d] pos_end[%d]", __PRETTY_FUNCTION__, __LINE__, fd, ret, errno, buf, pos_begin, pos_end);
			return (-1);
		}
		else {
//			LOG_DEBUG("%s %d %d: recv ret [%d] err [%d] buf[%p] pos_begin[%d] pos_end[%d]", __PRETTY_FUNCTION__, __LINE__, fd, ret, errno, buf, pos_begin, pos_end);
			return 2;
		}
	}
	else {
		pos_end += ret;
	}
	assert((int32_t)pos_end>=ret);
	return (0);
}

int conn_node_client::recv_handshake(int fd)
{
	int ret = recv_from_fd();
	if (ret != 0)
		return ret;
	int len = buf_size();

	char *end = (char *)buf_tail();
	if (len >= 4 && end[-1] == '\n' && end[-2] == '\r' && end[-3] == '\n' && end[-4] == '\r') {
		if (respond_websocket_request() != 0) //send websocket response
			return -1;
		// LOG_INFO("[%s : %d]: packet header error, len: %d, leave: %d", __PRETTY_FUNCTION__, __LINE__, len, buf_leave());
		pos_begin = pos_end = 0;
		handshake = true;
		return (0);
	}
	return (0);
}

int conn_node_client::get_listen_fd()
{
	return conn_node_client::listen_fd;
}

extern char *g_game_topic;
extern int   g_game_partid;

int conn_node_client::recv_func(int fd)
{
	int ret = -1;
	for (;;) {
		ret = recv(fd, buf_tail(), buf_leave(), 0);
		if (ret == 0)
		{
			LOG_INFO("%s %d %d: recv ret [%d] err [%d] buf[%p] pos_begin[%d] pos_end[%d]", __PRETTY_FUNCTION__, __LINE__, fd, ret, errno, buf, pos_begin, pos_end);
			return (-1);
		}
		else if (ret < 0)
		{
			if (errno != EAGAIN && errno != EINTR)
			{
				LOG_ERR("%s %d %d: recv ret [%d] err [%d] buf[%p] pos_begin[%d] pos_end[%d]", __PRETTY_FUNCTION__, __LINE__, fd, ret, errno, buf, pos_begin, pos_end);
				return (-1);
			}
			else
			{
				return 0;
			}
		}
		send_msg(buf_head(), ret, g_game_topic, g_game_partid);
		wakeup_send_thread(0);
		pos_begin = pos_end = 0;		
	}

	// PROTO_HEAD *head;
	// for (;;) {
	// 	int ret = get_one_buf();
	// 	if (ret == 0) {
	// 		head = (PROTO_HEAD *)buf_head();
	// 	}

	// 	if (ret < 0) {
	// 		LOG_INFO("%s: connect closed from fd %u, err = %d", __PRETTY_FUNCTION__, fd, errno);
	// 		return (-1);
	// 	} else if (ret > 0) {
	// 		break;
	// 	}

	// 	ret = remove_one_buf();
	// }
	return (0);
}

void encoder_data(PROTO_HEAD *head) {
/*
	size_t sz = ENDION_FUNC_4(head->len) - sizeof(uint16_t);
	if (sz <= 0)
		return;

	char* pData = (char*)head+sizeof(uint16_t);
	uint32_t crc32 = crc32_long((u_char*)head->data, ENDION_FUNC_4(head->len) - sizeof(PROTO_HEAD));
	head->crc = ENDION_FUNC_4(crc32);

	char * p = (char *)(pData);

	for (size_t i=0; i<sz/8; ++i) {
		sg_encrypt((uint32_t *)p);
		p += 8;
	}
*/
}

int conn_node_client::send_player_exit(bool again/* = false*/)
{
	return (0);
}

int conn_node_client::dispatch_message()
{
	return (0);
}

//////////////////////////// 下面是static 函数
std::map<int, conn_node_client *> conn_node_client::map_fd_nodes;
std::map<uint64_t, conn_node_client *> conn_node_client::map_player_id_nodes;
std::map<uint32_t, conn_node_client *> conn_node_client::map_open_id_nodes;
int conn_node_client::listen_fd;

conn_node_base *conn_node_client::get_conn_node(int fd)
{
	conn_node_client *ret = new conn_node_client;
	ret->buf = (uint8_t *)malloc(128 * 1024);
	ret->max_buf_len = 128 * 1024;
	ret->send_buffer = (char *)malloc(128 * 1024);
	ret->send_buffer_size = 128 * 1024;
	ret->open_id = 0;
	ret->player_id = 0;

	ret->handshake = false;
	ret->conn_step = 1;

	assert(map_fd_nodes.find(fd) == map_fd_nodes.end());
	map_fd_nodes[fd] = ret;
	return ret;
}

int conn_node_client::clean_node()
{
    if (player_id > 0)
    {
        map_player_id_nodes.erase(player_id);
//        write_to_shm_ipc(ipc_game_wr, NULL, 0, SERVER_MSG_OFFLINE, player_id);
    }
    if (open_id > 0)
        map_open_id_nodes.erase(open_id);
    player_id = 0;
    open_id   = 0;
    return (0);
}

int conn_node_client::del_node()
{
    assert(fd > 0);
    free(buf);
    free(send_buffer);

    LOG_DEBUG("client del: node = %p, fd = %u , player_id = %lu, open_id = %u", this, fd, player_id, open_id);

    map_fd_nodes.erase(fd);
    // if (player_id > 0) {
    //  map_player_id_nodes.erase(player_id);
    //  write_to_shm_ipc(ipc_game_wr, NULL, 0, SERVER_MSG_OFFLINE, player_id);
    // }
    // if (open_id > 0)
    //  map_open_id_nodes.erase(open_id);
    clean_node();

    fd = 0;
    conn_node_base::del_node();
    delete this;

    return (0);
}
int conn_node_client::add_map_fd_nodes(conn_node_client *client)
{
	map_fd_nodes[client->fd]=  client;
	return (0);
}
int conn_node_client::add_map_player_id_nodes(conn_node_client *client)
{
		//todo check duplicate
	if (client->player_id > 0)
		map_player_id_nodes[client->player_id] = client;
	return (0);
}

int conn_node_client::add_map_open_id_nodes(conn_node_client *client)
{
		//todo check duplicate
	if (client->open_id > 0)
		map_open_id_nodes[client->open_id] = client;
	return (0);
}

conn_node_client *conn_node_client::get_nodes_by_fd(int fd, uint16_t port)
{
	std::map<int, conn_node_client *>::iterator it = map_fd_nodes.find(fd);
	if (it != map_fd_nodes.end() && it->second->port == port)
		return it->second;
	return NULL;
}

conn_node_client *conn_node_client::get_nodes_by_player_id(uint64_t player_id)
{
	std::map<uint64_t, conn_node_client *>::iterator it = map_player_id_nodes.find(player_id);
	if (it != map_player_id_nodes.end())
		return it->second;
	return NULL;
}

conn_node_client *conn_node_client::get_nodes_by_open_id(uint32_t open_id)
{
	std::map<uint32_t, conn_node_client *>::iterator it = map_open_id_nodes.find(open_id);
	if (it != map_open_id_nodes.end())
		return it->second;
	return NULL;
}


